package com.junxonline.instant.admin.zookeeper;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.junxonline.instant.admin.config.ZookeeperConfig;
import com.junxonline.instant.dao.entity.dto.common.ServerNodeDTO;
import com.junxonline.instant.dao.entity.vo.monitor.ApplicationVO;
import lombok.extern.log4j.Log4j2;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.net.InetAddress;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;

/**
 * Zookeeper Client
 *
 * @author JunX
 * @date 2021-09-08
 */
@Log4j2
@Component
public class ZookeeperClient {

    @Autowired
    private ZookeeperConfig zookeeperConfig;

    /**
     * Zookeeper客户端
     */
    private CuratorFramework zookeeperClient;

    /**
     * ServerNode容器（线程安全）
     */
    // key="/services/xxx/xxx-server-0000000001"  value="192.168.2.17:7021"
    // key="/services/xxx/xxx-server-0000000002"  value="192.168.2.17:7022"
    // key="/services/xxx/xxx-server-0000000003"  value="192.168.2.17:7023"
    private final ConcurrentMap<String, String> serverNode = new ConcurrentHashMap<>();

    /**
     * Zookeeper Client启动
     *
     * @throws Exception Exception
     */
    @PostConstruct
    public void start() throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(zookeeperConfig.getBaseSleepTimeMs(), zookeeperConfig.getMaxRetries());
        zookeeperClient = CuratorFrameworkFactory
                .builder()
                .connectString(zookeeperConfig.getZookeeperAddress())
                .retryPolicy(retryPolicy)
                .sessionTimeoutMs(zookeeperConfig.getSessionTimeoutMs()).connectionTimeoutMs(zookeeperConfig.getConnectionTimeoutMs()).build();
        zookeeperClient.start();

        String name = zookeeperConfig.getId();
        String zookeeperAddress = zookeeperConfig.getZookeeperAddress();
        String ipPort = InetAddress.getLocalHost().getHostAddress() + ":" + zookeeperConfig.getPort();
        String data = name + "#" + zookeeperAddress + "#" + ipPort;
        zookeeperClient.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath(zookeeperConfig.getRegistryPath() + "/" + zookeeperConfig.getId() + "-", data.getBytes());
        log.info("- Zookeeper client started on -> {} in path: {}", ipPort, zookeeperConfig.getRegistryPath());

        // 对Zookeeper当中指定的Path进行监听
        CuratorCache curatorCache = CuratorCache.builder(zookeeperClient, zookeeperConfig.getListeningPath()).build();
        PathChildrenCacheListener pathChildrenCacheListener = (client, event) -> {
            event.getType();
            if (event.getType() == PathChildrenCacheEvent.Type.CONNECTION_RECONNECTED) {
                return;
            }
            log.info("Zookeeper event for basePath=[{}] event=[{}]", zookeeperConfig.getZookeeperAddress(), event);

            // 节点上线事件通知
            if (event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                String fullPath = event.getData().getPath();
                String eventData = new String(event.getData().getData(), StandardCharsets.UTF_8);
                eventData += "#" + event.getData().getStat().getCtime();
                log.info("Server node is online -> name: {}, info: {}", fullPath, eventData);
                serverNode.put(fullPath, eventData);
            }

            // 节点下线事件通知
            if (event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                String fullPath = event.getData().getPath();
                log.info("Server node is offline -> name: {}", fullPath);
                serverNode.remove(fullPath);
            }
        };
        CuratorCacheListener listener = CuratorCacheListener.builder().forPathChildrenCache(zookeeperConfig.getListeningPath(), zookeeperClient, pathChildrenCacheListener).build();
        curatorCache.listenable().addListener(listener);
        curatorCache.start();
        log.info("Zookeeper client listen on path -> {}", zookeeperConfig.getListeningPath());
    }


    /**
     * Zookeeper Client停止
     */
    @PreDestroy
    public void stop() {
        zookeeperClient.close();
        log.info("Zookeeper client stopped");
    }

    /**
     * 返回可用Server节点集合
     *
     * @param vo 请求参数
     * @return List<ServerNodeDTO>
     */
    public List<ServerNodeDTO> getServerNodes(ApplicationVO vo) {
        List<ServerNodeDTO> serverNodeDTOS = new ArrayList<>();
        int id = 1;
        for (ConcurrentMap.Entry<String, String> entry : serverNode.entrySet()) {
            String fullPath = entry.getKey();
            String data = entry.getValue();
            // 这里有可能得到空(读取到的是目录)
            if (StrUtil.isBlank(data)) {
                continue;
            }
            String[] dataArray = data.split("#");
            if (dataArray.length > 3) {
                String name = dataArray[0];
                String zookeeperAddress = dataArray[1];
                String url = dataArray[2];
                String startTime = dataArray[3];
                ServerNodeDTO dto = new ServerNodeDTO();
                dto.setId(id++);
                dto.setName(name);
                dto.setPath(fullPath);
                dto.setZookeeperAddress(zookeeperAddress);
                dto.setUrl(url);
                dto.setStartTime(DateUtil.date(Long.parseLong(startTime)).toString());
                serverNodeDTOS.add(dto);
            }
        }
        // 条件查询
        if (StrUtil.isNotBlank(vo.getName())) {
            String name = vo.getName();
            serverNodeDTOS = serverNodeDTOS.stream().filter(onlineUser -> onlineUser.getName().contains(name)).collect(Collectors.toList());
        }
        if (StrUtil.isNotBlank(vo.getPath())) {
            String path = vo.getPath();
            serverNodeDTOS = serverNodeDTOS.stream().filter(onlineUser -> onlineUser.getPath().contains(path)).collect(Collectors.toList());
        }
        if (StrUtil.isNotBlank(vo.getZookeeperAddress())) {
            String zookeeperAddress = vo.getZookeeperAddress();
            serverNodeDTOS = serverNodeDTOS.stream().filter(onlineUser -> onlineUser.getZookeeperAddress().contains(zookeeperAddress)).collect(Collectors.toList());
        }
        if (StrUtil.isNotBlank(vo.getUrl())) {
            String url = vo.getUrl();
            serverNodeDTOS = serverNodeDTOS.stream().filter(onlineUser -> onlineUser.getUrl().contains(url)).collect(Collectors.toList());
        }
        // 排序（按启动时间降序）
        serverNodeDTOS = serverNodeDTOS.stream().sorted(Comparator.comparing(ServerNodeDTO::getStartTime).reversed()).collect(Collectors.toList());
        return serverNodeDTOS;
    }

}
